#include "upload.h"

static struct shm_param para1, para2; // 共享内存参数
volatile MQTTClient_deliveryToken deliveredtoken;
static int *total_stm;          // 指向共享内存中数据节点总个数
static int *total_mod;          // 指向共享内存中Modbus节点总个数
static stm_node_p node_arr;     // 指向共享内存中节点缓存数组头
static modbus_set_p modbus_arr; // 指向共享内存中Modbus节点缓存数组头
uint16_t flush_flag = 0;        // 刷新标志
uint16_t send_type = 2;         // 发送类型
msgbuf_t send_buf;              // 消息缓冲区
uint16_t time_period = 2;       // 上报周期
MQTTClient client;
int history_pthread;
char his_report_buf[2048] = "";

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    deliveredtoken = dt; // 记录已发送消息的标识符
}

char *shm_get() // 获取共享内存数据
{
    int ret_stm = -1, ret_mod = -1;
    ret_stm = shm_init(&para1, "shm_stm32", MAX_NODE_SIZE); // 初始化stm共享内存
    ret_mod = shm_init(&para2, "shm_mod", MAX_NODE_SIZE);   // 初始化mod共享内存
    if (ret_stm < 0 || ret_mod < 0)
    {
        printf("Failed to initialize shared memory\n");
        return NULL;
    }
    char switch_buf[128];
    void *node_p_stm = shm_getaddr(&para1); // 获取stm共享内存地址
    void *node_p_mod = shm_getaddr(&para2); // 获取mod共享内存地址
    if (node_p_stm == NULL || node_p_mod == NULL)
    {
        printf("Failed to get shared memory address\n");
        return NULL;
    }

    total_stm = (int *)node_p_stm; // 前4个字节存储实际的数据节点总个数
    total_mod = (int *)node_p_mod;
    node_arr = (stm_node_p)(node_p_stm + sizeof(int));     // 后面空间存储stm数据点
    modbus_arr = (modbus_set_p)(node_p_mod + sizeof(int)); // 后面空间存储Modbus数据点
    printf("total_stm: %d, total_mod: %d\n", *total_stm, *total_mod);
    cJSON *root = cJSON_CreateObject(); // 创建根节点
    cJSON *data = cJSON_CreateArray();
    cJSON *num = cJSON_CreateNumber((*total_stm) + (*total_mod));
    cJSON_AddItemToObject(root, "total", num); // 添加总数

    for (int i = 0; i < (*total_stm); i++) // 遍历所有数据节点
    {
        cJSON *member = cJSON_CreateObject();
        cJSON *key = cJSON_CreateNumber(node_arr[i].key); // 读取dev_type值
        cJSON_AddItemToObject(member, "key", key);

        if (node_arr[i].type == TYPE_BOOL)
            snprintf(switch_buf, sizeof(switch_buf), "%s", node_arr[i].new_val.b_val ? "TRUE" : "FALSE");
        else if (node_arr[i].type == TYPE_FLOAT)
            snprintf(switch_buf, sizeof(switch_buf), "%.2f", node_arr[i].new_val.f_val);
        else if (node_arr[i].type == TYPE_INT)
            snprintf(switch_buf, sizeof(switch_buf), "%d", node_arr[i].new_val.i_val);

        cJSON *value = cJSON_CreateString(switch_buf);
        cJSON_AddItemToObject(member, "value", value);
        cJSON_AddItemToArray(data, member);
        memset(switch_buf, 0, sizeof(switch_buf));
    }
    for (int i = 0; i < (*total_mod); i++) // 遍历所有modbus数据节点
    {
        cJSON *member = cJSON_CreateObject();
        cJSON *key = cJSON_CreateNumber(modbus_arr[i].key); // 读取key值
        cJSON_AddItemToObject(member, "key", key);
        if (modbus_arr[i].type == TYPE_BOOL)
            snprintf(switch_buf, sizeof(switch_buf), "%s", modbus_arr[i].new_val.b_val ? "TRUE" : "FALSE");
        else if (modbus_arr[i].type == TYPE_FLOAT)
            snprintf(switch_buf, sizeof(switch_buf), "%.2f", modbus_arr[i].new_val.f_val);
        else if (modbus_arr[i].type == TYPE_INT)
            snprintf(switch_buf, sizeof(switch_buf), "%d", modbus_arr[i].new_val.i_val);
        cJSON *value = cJSON_CreateString(switch_buf); // 读取value值
        cJSON_AddItemToObject(member, "value", value);
        cJSON_AddItemToArray(data, member);
        memset(switch_buf, 0, sizeof(switch_buf));
    }
    cJSON_AddItemToObject(root, "data", data);
    if (!history_pthread)
    {
        pthread_t pid;
        pthread_create(&pid, NULL, (void *)history_data_upload, (void *)root); // 创建历史数据线程
        pthread_detach(pid);
    }

    char *p = cJSON_Print(root); // 获取JSON字符串
    // memset(node_p_mod, 0, MAX_NODE_SIZE);
    // memset(node_p_stm, 0, MAX_NODE_SIZE);
    return p;
}

void *MQTTClient_pub(void *client) // MQTTClient发布线程
{
    char payload_buf[64];
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer; // 初始化连接选项
    MQTTClient_message pubmsg = MQTTClient_message_initializer;                  // 初始化发布消息
    MQTTClient_deliveryToken token;
    MQTTClient *mqtt_client = (MQTTClient *)client;

    while (1)
    {
        int rc, i = 0;
        char *p = shm_get();
        printf("数据已更新\n");
        printf("send_type: %d\nflush_flag: %d\n", send_type, flush_flag);
        if (send_type == 0 && flush_flag)
        {
            printf("%s\n", p);
            pubmsg.payload = p;
            pubmsg.payloadlen = (int)strlen(p);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            flush_flag = 0; // 重置标志
            MQTTClient_publishMessage(mqtt_client, TOPIC1, &pubmsg, &token);
            rc = MQTTClient_waitForCompletion(mqtt_client, token, TIMEOUT);
            printf("--------刷新上报完成-------\n");
        }
        else if (send_type == 1)
        {
            for (; i < ((*total_mod) + (*total_stm)); i++)
            {
                if (i < (*total_stm) && node_arr[i].old_val.b_val != node_arr[i].new_val.b_val && node_arr[i].old_val.i_val != node_arr[i].new_val.i_val && node_arr[i].old_val.f_val != node_arr[i].new_val.f_val)
                {
                    switch (node_arr[i].type)
                    {
                    case TYPE_INT:
                        snprintf(payload_buf, sizeof(payload_buf), "%d", node_arr[i].new_val.i_val);
                        break;
                    case TYPE_BOOL:
                        snprintf(payload_buf, sizeof(payload_buf), "%s", node_arr[i].new_val.b_val ? "TRUE" : "FALSE");
                        break;
                    case TYPE_FLOAT:
                        snprintf(payload_buf, sizeof(payload_buf), "%.2f", node_arr[i].new_val.f_val);
                        break;
                    }
                }
                else if (i >= (*total_stm) && modbus_arr[i - (*total_stm)].old_val.b_val != modbus_arr[i - (*total_stm)].new_val.b_val && modbus_arr[i - (*total_stm)].old_val.i_val != modbus_arr[i - (*total_stm)].new_val.i_val && modbus_arr[i - (*total_stm)].old_val.f_val != modbus_arr[i - (*total_stm)].new_val.f_val)
                {
                    switch (modbus_arr[i - (*total_stm)].type)
                    {
                    case TYPE_INT:
                        snprintf(payload_buf, sizeof(payload_buf), "%d", modbus_arr[i - (*total_stm)].new_val.i_val);
                        break;
                    case TYPE_BOOL:
                        snprintf(payload_buf, sizeof(payload_buf), "%s", modbus_arr[i - (*total_stm)].new_val.b_val ? "TRUE" : "FALSE");
                        break;
                    case TYPE_FLOAT:
                        snprintf(payload_buf, sizeof(payload_buf), "%.2f", modbus_arr[i - (*total_stm)].new_val.f_val);
                        break;
                    }
                }
                pubmsg.payload = payload_buf;
                pubmsg.payloadlen = (int)strlen(pubmsg.payload);
                pubmsg.qos = QOS;
                pubmsg.retained = 0;
                printf("%s\n", payload_buf);
                if (strlen(payload_buf) > 0)
                {
                    MQTTClient_publishMessage(mqtt_client, TOPIC1, &pubmsg, &token);
                    rc = MQTTClient_waitForCompletion(mqtt_client, token, TIMEOUT);
                    memset(payload_buf, 0, sizeof(payload_buf));
                    printf("---------变化上报完成--------\n");
                }
                else
                    printf("---------无变化上报--------\n");
                    sleep(2);
            }
        }
        else if (send_type == 2)
        {
            printf("%s\n", p);
            pubmsg.payload = p;
            pubmsg.payloadlen = (int)strlen(p);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            MQTTClient_publishMessage(mqtt_client, TOPIC1, &pubmsg, &token);
            rc = MQTTClient_waitForCompletion(mqtt_client, token, TIMEOUT);
            printf("---------周期上报完成--------\n");
            sleep(time_period);
        }
        else if (send_type == 3)
        {
            if (strlen(his_report_buf) > 0)
            {
                pubmsg.payload = his_report_buf;
                pubmsg.payloadlen = (int)strlen(his_report_buf);
                pubmsg.qos = QOS;
                pubmsg.retained = 0;
                MQTTClient_publishMessage(mqtt_client, TOPIC1, &pubmsg, &token);
                rc = MQTTClient_waitForCompletion(mqtt_client, token, TIMEOUT);
                memset(his_report_buf, 0, sizeof(his_report_buf));
                printf("---------历史上报完成--------\n");
            }
            else
                printf("---------历史上报无数据--------\n");
        }
        free(p); // 释放由cJSON_Print分配的内存
    }
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

int main(int argc, char *argv[])
{
    int rc;
    // 定义连接选项变量
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;

    MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL); // 创建客户端实例
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);             // 设置回调函数

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Connection failed, code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    pthread_t tid;
    pthread_create(&tid, NULL, MQTTClient_pub, (void *)client);

    MQTTClient_subscribe(client, TOPIC2, QOS);
    pthread_join(tid, NULL); // 等待线程结束
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);

    return rc;
}